package com.atguigu.flink.wordcount;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

//静态导入。 把一个类的静态属性或方法，导入到当前类中。 导入完成后，就好比当前类编写了这些静态的方法或属性
import static org.apache.flink.api.common.typeinfo.Types.*;

/**
        如果是POJO，可以使用 returns(Class<T> typeClass)
        如果是Tuple，不能使用 returns(Class<T> typeClass) ，建议使用  returns(TypeHint)
 */
public class Demo7_LamdaTuple
{
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        DataStreamSource<String> ds = env.socketTextStream("hadoop103", 8888);

        //返回的数据作为key，会被分组
        ds.flatMap((FlatMapFunction<String, Tuple2<String, Integer>>) (value, out) -> {
            String[] words = value.split(" ");
            for (String word : words) {
                out.collect(Tuple2.of(word,1));
            }
        })
          //.returns(new TypeHint<Tuple2<String, Integer>>() {})
          //在flink提供了 Types，提供了现场的方法，帮你返回常见类型的TypeInformation对象
          .returns(TUPLE(STRING,INT))
          .keyBy( t ->   t.f0)
          .sum(1)
          .print();


        //启动执行环境
        env.execute();

    }
}
